package com.cars.controller;

import com.alibaba.fastjson2.JSON;
import com.cars.config.GlobalVariable.RedisKeyPrefix;
import com.cars.entity.Coffee;
import com.cars.repository.CoffeeRepository;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.domain.Sort;
import org.springframework.data.redis.core.ReactiveRedisOperations;
import org.springframework.web.bind.annotation.*;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

/**
 * @author: caoqianfan
 * @date: 2023/12/20 14:45
 */
@Slf4j
@RestController
@RequestMapping("/coffee")
public class CoffeeController {
    private static final Logger LOGGER = LoggerFactory.getLogger(CoffeeController.class);

    @Resource
    private ReactiveRedisOperations<String, Coffee> coffeeOps;
    @Resource
    CoffeeRepository repository;

    @PostMapping("/save")
    public Mono<Coffee> save(@RequestBody Coffee entity) {
        return repository.save(entity);
    }

    @DeleteMapping("/delete/{id}")
    public Mono<Void> delete(@PathVariable String id) {
        return repository.deleteById(id);
    }

    @PutMapping("/update")
    public Mono<Coffee> update(@RequestBody Coffee entity) {
        return repository.save(entity);
    }

    @GetMapping("/get/{id}")
    public Mono<Coffee> findById(@PathVariable Long id) {

        String redisKey = RedisKeyPrefix.coffee + id;
        Coffee query = new Coffee();
        query.setId(id);
        // 用于判断是否从redis获取到了缓存的对象
        List<Coffee> cached = new ArrayList<>();

        LOGGER.info("redis get start");

        Mono<Coffee> monoRedisGet = coffeeOps.opsForValue().get(redisKey).map(coffee -> {
            // 此段逻辑的处理线程，应与redis get start的线程不同
            LOGGER.info("redis get map");
            cached.add(coffee);
            return coffee;
        }).log();

        // 当redis里不存在缓存时，才到数据库查询
        Mono<Coffee> fluxQuery = monoRedisGet
                .switchIfEmpty(repository.findOne(org.springframework.data.domain.Example.of(query))).log();

        // 对Flux流的真正订阅，交由WebFlux框架负责
        return fluxQuery.flatMap(coffee -> {
            if (cached.size() == 0) {
                // 当redis里不存在缓存时，才写入缓存
                LOGGER.info("mysql query flatmap");
                return coffeeOps.opsForValue()
                        .set(redisKey, coffee, Duration.ofSeconds(10)).map(success -> {
                            // 此段逻辑的处理线程，应与mongo query flatmap的线程不同
                            LOGGER.info("redis set map");
                            return coffee;
                        });
            }
            return Mono.just(coffee);
        }).log();
    }

    @GetMapping("/list")
    public Flux<Coffee> list() {

        String redisKey = RedisKeyPrefix.coffee + "*";
        // 用于判断是否从redis获取到了缓存的对象
        List<Coffee> cached = new ArrayList<>();

        LOGGER.info("redis get start");

        Flux<Coffee> monoRedisGet = coffeeOps.keys(redisKey).map(str -> {
            LOGGER.warn(str);
            // 此段逻辑的处理线程，应与redis get start的线程不同
            Coffee coffee = JSON.parseObject(str, Coffee.class);
            LOGGER.info("redis get map");
            cached.add(coffee);
            return coffee;
        }).log();

        // 当redis里不存在缓存时，才到数据库查询
        Flux<Coffee> fluxQuery = monoRedisGet
                .switchIfEmpty(repository.findAll(Sort.by(Sort.Direction.ASC, "id"))).log();

        // 对Flux流的真正订阅，交由WebFlux框架负责
        return fluxQuery.flatMap(coffee -> {
            if (cached.size() == 0) {
                // 当redis里不存在缓存时，才写入缓存
                LOGGER.info("mysql query flatmap");
                return coffeeOps.opsForValue()
                        .set(RedisKeyPrefix.coffee + coffee.getId(), coffee, Duration.ofSeconds(10)).map(success -> {
                            // 此段逻辑的处理线程，应与mongo query flatmap的线程不同
                            LOGGER.info("redis set map");
                            return coffee;
                        });
            }
            return Flux.just(coffee);
        }).log();
    }
}